package com.mqstudy.kafka.controller;

import com.mqstudy.common.model.Message;
import com.mqstudy.common.util.MessageUtil;
import com.mqstudy.kafka.producer.KafkaProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.HashMap;
import java.util.Map;

/**
 * Kafka控制器
 */
@Slf4j
@RestController
@RequestMapping("/kafka")
public class KafkaController {
    
    @Autowired
    private KafkaProducer producer;
    
    /**
     * 发送测试消息
     */
    @PostMapping("/send/test")
    public Map<String, Object> sendTestMessage(@RequestBody Map<String, String> request) {
        try {
            String content = request.get("content");
            Message message = MessageUtil.createTestMessage("test-topic", content);
            producer.sendTestMessage(message);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "测试消息发送成功");
            result.put("messageId", message.getMessageId());
            return result;
            
        } catch (Exception e) {
            log.error("发送测试消息失败: {}", e.getMessage(), e);
            Map<String, Object> result = new HashMap<>();
            result.put("success", false);
            result.put("message", "发送失败: " + e.getMessage());
            return result;
        }
    }
    
    /**
     * 发送订单消息
     */
    @PostMapping("/send/order")
    public Map<String, Object> sendOrderMessage(@RequestBody Map<String, String> request) {
        try {
            String orderId = request.get("orderId");
            String orderInfo = request.get("orderInfo");
            Message message = MessageUtil.createOrderMessage(orderId, orderInfo);
            producer.sendOrderMessage(message);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "订单消息发送成功");
            result.put("messageId", message.getMessageId());
            return result;
            
        } catch (Exception e) {
            log.error("发送订单消息失败: {}", e.getMessage(), e);
            Map<String, Object> result = new HashMap<>();
            result.put("success", false);
            result.put("message", "发送失败: " + e.getMessage());
            return result;
        }
    }
    
    /**
     * 发送用户消息
     */
    @PostMapping("/send/user")
    public Map<String, Object> sendUserMessage(@RequestBody Map<String, String> request) {
        try {
            String content = request.get("content");
            Message message = MessageUtil.createTestMessage("user-topic", content);
            producer.sendUserMessage(message);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "用户消息发送成功");
            result.put("messageId", message.getMessageId());
            return result;
            
        } catch (Exception e) {
            log.error("发送用户消息失败: {}", e.getMessage(), e);
            Map<String, Object> result = new HashMap<>();
            result.put("success", false);
            result.put("message", "发送失败: " + e.getMessage());
            return result;
        }
    }
    
    /**
     * 发送日志消息
     */
    @PostMapping("/send/log")
    public Map<String, Object> sendLogMessage(@RequestBody Map<String, String> request) {
        try {
            String content = request.get("content");
            Message message = MessageUtil.createTestMessage("log-topic", content);
            producer.sendLogMessage(message);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "日志消息发送成功");
            result.put("messageId", message.getMessageId());
            return result;
            
        } catch (Exception e) {
            log.error("发送日志消息失败: {}", e.getMessage(), e);
            Map<String, Object> result = new HashMap<>();
            result.put("success", false);
            result.put("message", "发送失败: " + e.getMessage());
            return result;
        }
    }
    
    /**
     * 发送流式消息
     */
    @PostMapping("/send/stream")
    public Map<String, Object> sendStreamMessage(@RequestBody Map<String, String> request) {
        try {
            String content = request.get("content");
            Message message = MessageUtil.createTestMessage("stream-topic", content);
            producer.sendStreamMessage(message);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "流式消息发送成功");
            result.put("messageId", message.getMessageId());
            return result;
            
        } catch (Exception e) {
            log.error("发送流式消息失败: {}", e.getMessage(), e);
            Map<String, Object> result = new HashMap<>();
            result.put("success", false);
            result.put("message", "发送失败: " + e.getMessage());
            return result;
        }
    }
    
    /**
     * 发送批量消息
     */
    @PostMapping("/send/batch")
    public Map<String, Object> sendBatchMessages(@RequestBody Map<String, Object> request) {
        try {
            String topic = (String) request.get("topic");
            Integer count = (Integer) request.get("count");
            
            if (topic == null) {
                topic = "batch-topic";
            }
            if (count == null || count <= 0) {
                count = 10;
            }
            
            producer.sendBatchMessages(topic, count);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "批量消息发送成功，数量: " + count);
            result.put("topic", topic);
            result.put("count", count);
            return result;
            
        } catch (Exception e) {
            log.error("发送批量消息失败: {}", e.getMessage(), e);
            Map<String, Object> result = new HashMap<>();
            result.put("success", false);
            result.put("message", "发送失败: " + e.getMessage());
            return result;
        }
    }
    
    /**
     * 健康检查
     */
    @GetMapping("/health")
    public Map<String, Object> health() {
        Map<String, Object> result = new HashMap<>();
        result.put("status", "UP");
        result.put("service", "Kafka Demo");
        result.put("port", 8082);
        return result;
    }
} 